package com.kafka.demo.consumer.cluster;

import com.kafka.demo.cons.KafkaTopic;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * 集群消费者-重试+手动提交
 */
@Component
@Slf4j
public class RetryConsumer {

    // ✅ 使用 ThreadLocal 记录当前消息的重试次数
    private static final ThreadLocal<Integer> RETRY_COUNT = ThreadLocal.withInitial(() -> 0);

    @KafkaListener(
            id = "cluster-consumer2",
            topics = KafkaTopic.MY_TOPIC,
            group = "${kafka-consumer.services.cluster-consumer2.group-id}",
            containerFactory = "clusterListenerFactory2"
    )
    public void listen(ConsumerRecord<String, String> record, Acknowledgment ack) {
        try {
            // 获取当前重试次数
            int currentRetry = RETRY_COUNT.get();
            log.info("收到消息: Partition={}, Offset={}, Key={}, Value={}, 重试次数={}",
                    record.partition(), record.offset(), record.key(), record.value(), currentRetry);

            // 模拟业务异常
            int i = 1 / 0;

        } catch (Exception e) {
            int currentRetry = RETRY_COUNT.get();
            log.warn("消息处理失败，Offset={}, 重试次数={}", record.offset(), currentRetry, e);

            // ✅ 判断是否达到最大重试次数（3次）
            if (currentRetry >= 2) {
                // 第3次（第0、1、2次）→ 最后一次，手动确认，避免无限重试
                log.warn("达到最大重试次数，手动确认并放弃处理，Offset={}", record.offset());
                ack.acknowledge(); // ✅ 手动提交 offset，跳过这条消息
                RETRY_COUNT.remove();
            }
            RETRY_COUNT.set(RETRY_COUNT.get() + 1);
            // ❗ 无论是否达到最大重试，都抛出异常 → 触发 RetryTemplate 的 backoff 控制
            // 但最后一次我们已经 ack，即使重试也不会再消费（因为 offset 提交了）
            throw e; // 触发重试机制

        } finally {
        }
    }
}